package com.viknix.threadpool.manager.cluster.file;

import cn.hutool.core.collection.ConcurrentHashSet;
import com.viknix.threadpool.manager.server.dependence.factory.ExecutorFactory;
import com.viknix.threadpool.manager.server.dependence.factory.NameThreadFactory;
import com.viknix.threadpool.manager.server.dependence.exception.ServerProcessException;
import com.viknix.threadpool.manager.server.dependence.util.ThreadUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.nio.file.*;
import java.util.*;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @Author: Dongqi
 * @Date: 2021/11/11 18:29
 * @Version 1.0
 * @Description: 统一文件变更监控管理中心，内部使用{@link WatchService}。
 * 一个文件目录对应一个{@link WatchService}。它最多只能监视 32 个文件目录。
 * 当发生文件更改时，将发出{@link FileChangeEvent}
 */
public class WatchFileCenter {

    private static final Logger LOGGER = LoggerFactory.getLogger(WatchFileCenter.class);

    /**
     * 受监控文件目录的最大数量。
     */
    private static final int MAX_WATCH_FILE_JOB = Integer.getInteger("tpm.watch-file.max-dirs", 16);

    /**
     * key:paths, value:WatchDirJob
     * 用于存放被监控的目录所对应的监控任务 {@link WatchDirJob}
     */
    private static final Map<String, WatchDirJob> MANAGER = new HashMap<String, WatchDirJob>(MAX_WATCH_FILE_JOB);

    private static final FileSystem FILE_SYSTEM = FileSystems.getDefault();

    private static final AtomicBoolean CLOSED = new AtomicBoolean(false);

    static {
        ThreadUtils.addShutdownHook(new Runnable() {
            @Override
            public void run() {
                shutdown();
            }
        });
    }

    /**
     * 当前监视的目录数。
     */
    @SuppressWarnings("checkstyle:StaticVariableName")
    private static int NOW_WATCH_JOB_CNT = 0;

    /**
     * 在此目录中注册 {@link FileWatcher}。
     * 当该目录下文件发生更新时，会回调 watcher
     *
     * @param paths   需要被监控的目录
     * @param watcher {@link FileWatcher}
     * @return 注册是否成功
     * @throws ServerProcessException serverProcessException
     */
    public static synchronized boolean registerWatcher(final String paths, FileWatcher watcher) throws ServerProcessException {
        // 判断该文件变更系统是否系统关闭
        checkState();
        // 如果当前监控的目录数已经等于该文件系统所能监控的最大目录数，监控失败，返回 false
        if (NOW_WATCH_JOB_CNT == MAX_WATCH_FILE_JOB) {
            return false;
        }
        // 返回当前目录对应的 WatchDirJob，如果为空则重新构建一个交给 MANAGER集合 进行管理
        WatchDirJob job = MANAGER.get(paths);
        if (job == null) {
            job = new WatchDirJob(paths);
            job.start();
            MANAGER.put(paths, job);
            // 当前监视的目录数递增
            NOW_WATCH_JOB_CNT++;
        }
        // 将 watcher 注册到 WatchDirJob 中，方便该目录发生变更时进行回调
        job.addSubscribe(watcher);
        return true;
    }

    /**
     * 注销该目录中的所有监控 {@link FileWatcher}。
     *
     * @param path 目录路径
     * @return 注销成功
     */
    public static synchronized boolean deregisterAllWatcher(final String path) {
        WatchDirJob job = MANAGER.get(path);
        if (job != null) {
            job.shutdown();
            // 从 MANAGER 中移除监控
            MANAGER.remove(path);
            NOW_WATCH_JOB_CNT--;
            return true;
        }
        return false;
    }

    /**
     * 关闭 {@link WatchFileCenter}.
     */
    public static void shutdown() {
        if (!CLOSED.compareAndSet(false, true)) {
            return;
        }
        LOGGER.warn("[WatchFileCenter] start close");
        // 依次关闭监控任务
        for (Map.Entry<String, WatchDirJob> entry : MANAGER.entrySet()) {
            LOGGER.warn("[WatchFileCenter] start to shutdown this watcher which is watch : " + entry.getKey());
            try {
                entry.getValue().shutdown();
            } catch (Throwable e) {
                LOGGER.error("[WatchFileCenter] shutdown has error : {}", e);
            }
        }
        // 清空 MANAGER
        MANAGER.clear();
        NOW_WATCH_JOB_CNT = 0;
        LOGGER.warn("[WatchFileCenter] already closed");
    }

    /**
     * 注销此目录中指定的 FileWatcher
     *
     * @param path    目录路径
     * @param watcher 需要注销的 {@link FileWatcher}
     * @return 注销是否成功
     */
    public static synchronized boolean deregisterWatcher(final String path, final FileWatcher watcher) {
        WatchDirJob job = MANAGER.get(path);
        if (job != null) {
            job.watchers.remove(watcher);
            return true;
        }
        return false;
    }

    /**
     * 用于执行监控任务的线程
     */
    private static class WatchDirJob extends Thread {

        /**
         * 用于回调 watch 的线程池对象
         */
        private ExecutorService callBackExecutor;

        /**
         * 该监控任务监控的目录路径
         */
        private final String paths;

        private WatchService watchService;

        private volatile boolean watch = true;

        /**
         * 当前文件目录发生事件时需要回调的 watchers
         */
        private Set<FileWatcher> watchers = new ConcurrentHashSet<>();

        public WatchDirJob(String paths) throws ServerProcessException {
            setName(paths);
            this.paths = paths;
            final Path p = Paths.get(paths);
            if (!p.toFile().isDirectory()) {
                throw new IllegalArgumentException("Must be a file directory : " + paths);
            }

            // 用于回调 watch 的线程池对象
            this.callBackExecutor = ExecutorFactory
                    .newSingleExecutorService(new NameThreadFactory("com.viknix.tpm.sys.file.watch-" + paths));

            try {
                // 构建一个新的 WatchService 对象用于监控当前 paths
                WatchService service = FILE_SYSTEM.newWatchService();
                // 将当前 paths 注册到 WatchService 上，并设置关注的事件
                p.register(service, StandardWatchEventKinds.OVERFLOW, StandardWatchEventKinds.ENTRY_MODIFY,
                        StandardWatchEventKinds.ENTRY_CREATE, StandardWatchEventKinds.ENTRY_DELETE);
                this.watchService = service;
            } catch (Throwable e) {
                throw new ServerProcessException(ServerProcessException.SERVER_ERROR, e);
            }
        }

        /**
         * 将 watcher 注册到 WatchDirJob 中，方便该目录发生变更时进行回调
         *
         * @param watcher {@link FileWatcher}
         */
        void addSubscribe(final FileWatcher watcher) {
            watchers.add(watcher);
        }

        /**
         * 关闭当前监控线程，并释放资源
         */
        void shutdown() {
            watch = false;
            ThreadUtils.shutdownThreadPool(this.callBackExecutor);
        }

        @Override
        public void run() {
            while (watch) {
                try {
                    // 等待当前文件目录的事件发生
                    final WatchKey watchKey = watchService.take();
                    // 获取所有事件
                    final List<WatchEvent<?>> events = watchKey.pollEvents();
                    // 重置监视键
                    watchKey.reset();
                    if (callBackExecutor.isShutdown()) {
                        return;
                    }
                    if (events.isEmpty()) {
                        continue;
                    }
                    callBackExecutor.execute(new Runnable() {
                        @Override
                        public void run() {
                            for (WatchEvent<?> event : events) {
                                WatchEvent.Kind<?> kind = event.kind();

                                // Since the OS's event cache may be overflow, a backstop is needed
                                if (StandardWatchEventKinds.OVERFLOW.equals(kind)) {
                                    eventOverflow();
                                } else {
                                    // 构建 FileChangeEvent 事件，并回调 FileWatcher
                                    eventProcess(event.context());
                                }
                            }
                        }
                    });
                } catch (InterruptedException ignore) {
                    Thread.interrupted();
                } catch (Throwable ex) {
                    LOGGER.error("An exception occurred during file listening : {}", ex);
                }
            }
        }

        /**
         * 构建 FileChangeEvent 事件，并回调 FileWatcher
         *
         * @param context 变更上下文
         */
        private void eventProcess(Object context) {
            // 构建 FileChangeEvent 事件
            final FileChangeEvent fileChangeEvent = FileChangeEvent.builder().paths(paths).context(context).build();
            final String str = String.valueOf(context);
            for (final FileWatcher watcher : watchers) {
                // 判断该 watcher 是否对当前变更感兴趣
                if (watcher.interest(str)) {
                    // 构建回调任务
                    Runnable job = new Runnable() {
                        @Override
                        public void run() {
                            watcher.onChange(fileChangeEvent);
                        }
                    };
                    // 如果当前 watcher 已经内置了线程池，直接交给内置的线程池进行处理
                    // 如果没有内置线程池，由当前线程执行回调任务
                    Executor executor = watcher.executor();
                    if (executor == null) {
                        try {
                            job.run();
                        } catch (Throwable ex) {
                            LOGGER.error("File change event callback error : {}", ex);
                        }
                    } else {
                        executor.execute(job);
                    }
                }
            }
        }

        private void eventOverflow() {
            File dir = Paths.get(paths).toFile();
            for (File file : Objects.requireNonNull(dir.listFiles())) {
                // Subdirectories do not participate in listening
                if (file.isDirectory()) {
                    continue;
                }
                eventProcess(file.getName());
            }
        }

    }

    /**
     * 判断该文件变更系统是否系统关闭
     */
    private static void checkState() {
        if (CLOSED.get()) {
            throw new IllegalStateException("WatchFileCenter already shutdown");
        }
    }
}
